package org.example.releaseConfirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import org.example.utils.RabbitMqUtils;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author zfj
 * @date 2022/4/23 -13:28
 *
 * 发布确认模式
 * 单个确认模式   //发布1000个单独确认消息耗时1184ms
 * 批量确认模式   发布1000个批量确认消息耗时144ms
 * 异步批量确认模式  发布1000个异步发布确认消息耗时88ms
 * 使用的时间来比较哪种方式是最好的
 */
public class ConfirmMessage {
	//批量发消息的个数
	public static final int MESSAGE_COUNT = 1000;

	public static void main(String[] args) throws Exception {
		//单个确认模式
		//ConfirmMessage.publishMessageIndividually();
		//批量确认模式
//		ConfirmMessage.publishMessageBatch();
		//异步批量确认模式
		ConfirmMessage.publishMessageAsync();
	}
	//单个确认
	public static void publishMessageIndividually() throws Exception{
		Channel channel = RabbitMqUtils.getChannel();
		//队列声明
		String queueName = UUID.randomUUID().toString();
		channel.queueDeclare(queueName,false,false,false,null);
		//开启发布确认
		channel.confirmSelect();
		//开始时间
		long begin = System.currentTimeMillis();
		//批量发消息
		for (int i = 0; i < MESSAGE_COUNT; i++) {
			String message=i+"";
			channel.basicPublish("",queueName,null,message.getBytes());
			//单个消息马上发布确认
			boolean flag= channel.waitForConfirms();
			if (flag){
				System.out.println("消息发送成功");
			}
		}
		//结束时间
		long end = System.currentTimeMillis();
		System.out.println("发布"+MESSAGE_COUNT+"个单独确认消息耗时"+(end - begin)+"ms");
	}

	//批量确认模式
	public static void publishMessageBatch() throws Exception{
		Channel channel = RabbitMqUtils.getChannel();
		//队列声明
		String queueName = UUID.randomUUID().toString();
		channel.queueDeclare(queueName,false,false,false,null);
		//开启发布确认
		channel.confirmSelect();
		//开始时间
		long begin = System.currentTimeMillis();

		//批量却仍消息大小
		int batchSize=100;
		//批量发消息
		for (int i = 0; i < MESSAGE_COUNT; i++) {
			String message=i+"";
			channel.basicPublish("",queueName,null,message.getBytes());
			//判断达到100条批量确认一次
			if(i%batchSize==0){
				//发布确认
				channel.waitForConfirms();
			}
		}
		//结束时间
		long end = System.currentTimeMillis();
		System.out.println("发布"+MESSAGE_COUNT+"个批量确认消息耗时"+(end - begin)+"ms");
	}

	//异步发布确认模式
	public static void publishMessageAsync() throws Exception{
		Channel channel = RabbitMqUtils.getChannel();
		//队列声明
		String queueName = UUID.randomUUID().toString();
		channel.queueDeclare(queueName,false,false,false,null);
		//开启发布确认
		channel.confirmSelect();
		/**
		 * 线程安全有序的哈希表，适用于高并发的情况下
		 * 1.轻松的将序号与消息关联
		 * 2.轻松批量删除条数，只要给到序号
		 * 3.支持高并发
		 */
		ConcurrentSkipListMap<Long,String> outstandingConfirms=
				new ConcurrentSkipListMap<>();
		//准备消息的监听器，监听哪些消息发送成功，哪些失败了
		//消息确认成功回调
		ConfirmCallback ackCallback=(deliveryTag, multiple)->{
			//2.删除掉已经确认消息，剩下的就是未确认的信息
			if(multiple){
				ConcurrentNavigableMap<Long, String> confirmed= outstandingConfirms.headMap(deliveryTag);
				confirmed.clear();
			}else {
				outstandingConfirms.remove(deliveryTag);
			}
			System.out.println("确认的消息："+deliveryTag);
		};
		//消息确认失败回调
		/**
		 * 1.消息的标记
		 * 2.是否为批量操作
		 */
		ConfirmCallback nackCallback=(deliveryTag, multiple)->{
			//3.打印下来未确认的消息有哪些
			String message = outstandingConfirms.get(deliveryTag);
			System.out.println("未确认的消息："+message+"未确认的消息："+deliveryTag);
		};
		/**
		 * 1.监听成功
		 * 2.监听失败
		 */
		channel.addConfirmListener(ackCallback,nackCallback);//异步通知
		//开始时间
		long begin = System.currentTimeMillis();
		//批量发消息
		for (int i = 0; i < MESSAGE_COUNT; i++) {
			String message=i+"";
			channel.basicPublish("",queueName,null,message.getBytes());
			//1.此处记录下所有要发送的消息，消息的总和
			outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
		}

		//结束时间
		long end = System.currentTimeMillis();
		System.out.println("发布"+MESSAGE_COUNT+"个异步发布确认消息耗时"+(end - begin)+"ms");
	}

}
